package cn.tiansu.eway.kylinengine.manager;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import cn.tiansu.eway.common.utils.PropertyConfigurator;
import cn.tiansu.eway.kafka.ijava.ConsumerConsumption;
import cn.tiansu.eway.kafka.ijava.TSMessageQueue;
import cn.tiansu.eway.kylinengine.service.KYLINEngineService;
import cn.tiansu.eway.module.RunningStatistic;

import com.alibaba.fastjson.JSONObject;

/**
 * 
 * @Title: KYLINEngineListener.java
 * @Description: kylin引擎监听器
 * @author: 吴鹏
 * @date: 2015年8月13日
 * 
 *
 */
public class KYLINEngineListener extends ConsumerConsumption
{
	/**
	 * log
	 */
	private final static Logger log = Logger.getLogger(KYLINEngineListener.class);

	private final String PubModuleStatus = "PubModuleStatus";

	private TSMessageQueue teQueue = new TSMessageQueue(this);

	/**
	 * olap和kylin的交互模块
	 */
	private KYLINEngineService kylinService = KYLINEngineService.getInstance();

	private ThreadPoolExecutor buildExecutor = new ThreadPoolExecutor(10, 200,
			20000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(50));

	private ThreadPoolExecutor messageExecutor = new ThreadPoolExecutor(5, 10,
			20000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(10));

	
	/**
	 * 租户set
	 */
	public static Set<String> tenantSet = new HashSet<String>();

	// olap接口发送消息长度
	private int sendLen = 0;

	// olap接口接收消息长度
	private int recLen = 0;

	public KYLINEngineListener()
	{
		teQueue.open(KYLINManager.MODELNAME, KYLINManager.CMD);
		String[] arr =
		{ KYLINManager.CMD, KYLINManager.tenantUpdate};
		teQueue.subscibeMsg(arr);
		teQueue.start();
	}

	public void init()
	{
		tenantSet = kylinService.getTenantName();
	}

	/**
	 * 收到消息后执行对应的业务逻辑
	 * 
	 * @message 消息
	 */
	@Override
	public void consumptionMsg(String message)
	{
		log.info("message is "+ message);
		
		if (message.indexOf("olaps") > -1)
		{
			try
			{
				JSONObject result = kylinService.queryDataByCondition(message);
				String res = teQueue.responseMsg(message, 0, result);
				log.info("answer msg is :" + res);
			} catch (Exception e)
			{
				log.error("consumptionMsg is fail !", e);
			}
		}
		else if (message.indexOf("msg") > -1)
		{
			kylinService.updateTenant(message);
		} 

//		 MessageThread t = new MessageThread(message, kylinService, teQueue);
//		 messageExecutor.submit(t);
	}

	/**
	 * 周期性的重建cube
	 */
	public void buildCubePeriod()
	{
		PropertyConfigurator config = PropertyConfigurator.getInstance(null,
				"config.properties");

		// 单位为秒
		long period = Long.valueOf(config.getParameter("buildPeriod"));

		log.info("目前有租户："+tenantSet);

		while (true)
		{
			// ==========
//			try
//			{
//				log.info("========开始休眠==========");
//				Thread.sleep(period * 1000);
//			} catch (InterruptedException e)
//			{
//				log.error(e);
//			}
			// ==========

			log.info(" first clean kylin's cache");

			Iterator<String> it = tenantSet.iterator();

			kylinService.reloadMeta();

			while (it.hasNext())
			{
				// 构建新立方
				MutilThread thread = new MutilThread(it.next(), kylinService,period);
				buildExecutor.submit(thread);
			}

			// ==== test begin
			try
			{
				log.info("========开始休眠==========");
				Thread.sleep(period * 1000);
			} catch (InterruptedException e)
			{
				log.error(e);
			}
			// ==== test end
		}
	}

	// 内部测试代码，用来测试使用
	public String sendMessageTest(String str)
	{
		JSONObject message = JSONObject.parseObject(str);
		String res = teQueue.requestMsg(KYLINManager.MODELNAME,
				KYLINManager.CMD, message, 0);
		System.out.println(res);
		return res;
	}

	/**
	 * 发送模块统计数据，包括cpu，内存，运行时间等信息
	 * 
	 * @param modelName
	 *            模块名称
	 * @param privateName
	 *            私有名称
	 */
	public void sendPubModuleStatus(String modelName, String privateName)
	{
		new Thread(new MyTask(modelName, privateName)).start();
	}

	/**
	 * 内部类，用来定时发送运行状态
	 * 
	 * @author user
	 *
	 */
	class MyTask implements Runnable
	{
		String modelName = null;
		String privateName = null;

		public MyTask(String modelName, String privateName)
		{
			this.modelName = modelName;
			this.privateName = privateName;
		}

		@Override
		public void run()
		{
			PropertyConfigurator config = PropertyConfigurator.getInstance(
					null, "config.properties");

			RunningStatistic runningStatistic = RunningStatistic.factory();
			runningStatistic.runningInfoInit(modelName, privateName,
					config.getParameter("verion"));

			String period = config.getParameter("runningStatusPeriod");
			period = (period == null) ? "10" : period;

			while (true)
			{
				runningStatistic.runningStatistic(String.valueOf(sendLen),
						String.valueOf(recLen));
				sendLen = 0;
				recLen = 0;
				JSONObject jsonObject = runningStatistic.buildRunningInfoJSON();

				teQueue.publishMsg(PubModuleStatus, jsonObject, 0);
				log.debug("runningInfoInit is "+ jsonObject.toJSONString());

				try
				{
					Thread.sleep(Long.valueOf(period));

				} catch (InterruptedException e)
				{
					log.error("", e);
				}
			}
		}
	}
}
